package com.example.springbootbatch.config;

import com.example.springbootbatch.domain.Person;
import com.example.springbootbatch.listener.CsvJobListener;
import com.example.springbootbatch.validator.CsvBeanValidator;
import com.example.springbootbatch.validator.CsvItemProcessor;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
import org.springframework.batch.core.launch.support.SimpleJobLauncher;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.repository.support.JobRepositoryFactoryBean;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.database.BeanPropertyItemSqlParameterSourceProvider;
import org.springframework.batch.item.database.JdbcBatchItemWriter;
import org.springframework.batch.item.file.FlatFileItemReader;
import org.springframework.batch.item.file.mapping.BeanWrapperFieldSetMapper;
import org.springframework.batch.item.file.mapping.DefaultLineMapper;
import org.springframework.batch.item.file.transform.DelimitedLineTokenizer;
import org.springframework.batch.item.validator.Validator;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.ClassPathResource;
import org.springframework.transaction.PlatformTransactionManager;

import javax.sql.DataSource;

/**
 * 配置spring.batch.job.enable=false
 * 手动触发 Job 监听器执行任务
 * @author ljw
 * @date 2022/8/31 8:07
 */
@Configuration
@EnableBatchProcessing
public class TriggerBatchConfig {


    /**
     * Job 监听
     * @return
     */
    @Bean
    public CsvJobListener csvJobListener() {
        return new CsvJobListener();
    }

    /**
     * 自定义校验器
     * @return
     */
    @Bean
    public Validator<Person> csvBeanValidator() {
        return new CsvBeanValidator<>();
    }

    /**
     * 读取数据的接口
     * 参数后置绑定，在 JobParameters 中绑定参数，在 Bean 定义的时候使用一个特殊的 Bean 生命周期注解 @StepScope, 然后通过 @Value 注入此参数
     */
    @Bean
    @StepScope
    public FlatFileItemReader<Person> reader(@Value("#{jobParameters['input.file.name']}") String pathToFile) throws Exception {
        //FlatFileItemReader 读取文件
        FlatFileItemReader<Person> reader = new FlatFileItemReader<>();
        //setResource 方法设置 csv 文件的路径
        reader.setResource(new ClassPathResource(pathToFile));
        //在此处对 cvs 文件的数据和领域模型做对应映射
        reader.setLineMapper(new DefaultLineMapper<Person>(){{
            setLineTokenizer(new DelimitedLineTokenizer(){{
                setNames(new String[]{"name", "age", "nation", "address"});
            }});
            setFieldSetMapper(new BeanWrapperFieldSetMapper<Person>(){{
                setTargetType(Person.class);
            }});
        }});
        return reader;
    }

    /**
     * 数据处理、数据校验
     * @return
     */
    @Bean
    public ItemProcessor<Person, Person> processor() {
        //自定义的 ItemProcessor
        CsvItemProcessor processor = new CsvItemProcessor();
        //为 processor 指定校验器
        processor.setValidator(csvBeanValidator());
        return processor;
    }

    /**
     * 输出数据的接口
     * @param dataSource
     * @return
     */
    @Bean
    public ItemWriter<Person> writer(DataSource dataSource) {
        //使用 JDBC 批处理的 JdbcBatchItemWriter 来写数据到数据库
        JdbcBatchItemWriter<Person> writer = new JdbcBatchItemWriter<>();
        writer.setItemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider<Person>());
        String sql = "insert into person "+ "(id, name, age, nation, address) " + "values(:id, :name, :age, :nation, :address)";
        //设置要执行批处理的 SQL 语句
        writer.setSql(sql);
        writer.setDataSource(dataSource);
        return writer;
    }

    /**
     * 注册 Job 的容器
     * @param dataSource
     * @param transactionManager
     * @return
     * @throws Exception
     */
    @Bean
    public JobRepository jobRepository(DataSource dataSource, PlatformTransactionManager transactionManager) throws Exception {
        JobRepositoryFactoryBean jobRepositoryFactoryBean = new JobRepositoryFactoryBean();
        jobRepositoryFactoryBean.setDataSource(dataSource);
        jobRepositoryFactoryBean.setTransactionManager(transactionManager);
        jobRepositoryFactoryBean.setDatabaseType("mysql");
        return jobRepositoryFactoryBean.getObject();
    }

    /**
     * 启动 Job 的接口
     * @param dataSource
     * @param transactionManager
     * @return
     */
    @Bean
    public SimpleJobLauncher jobLauncher(DataSource dataSource, PlatformTransactionManager transactionManager) throws Exception {
        SimpleJobLauncher jobLauncher = new SimpleJobLauncher();
        jobLauncher.setJobRepository(jobRepository(dataSource, transactionManager));
        return jobLauncher;
    }

    /**
     * 注册并绑定监听器到 Job
     * @param jobs
     * @param step1
     * @return
     */
    @Bean
    public Job importJob(JobBuilderFactory jobs, Step step1) {
        return jobs.get("importJob")
                .incrementer(new RunIdIncrementer())
                //为 Job 指定 Step
                .flow(step1)
                .end()
                //绑定监听器
                .listener(csvJobListener())
                .build();
    }

    /**
     * 步骤，包含 ItemReader,ItemProcessor 和 ItemWrter
     * @param stepBuilderFactory
     * @param reader
     * @param writer
     * @param processor
     * @return
     */
    @Bean
    public Step step1(StepBuilderFactory stepBuilderFactory, ItemReader<Person> reader, ItemWriter<Person> writer, ItemProcessor<Person, Person> processor) {
        return stepBuilderFactory.get("step1")
                //批处理每次提交65000条数据
                .<Person, Person> chunk(65000)
                //给 step 绑定reader、processor、writer
                .reader(reader)
                .processor(processor)
                .writer(writer)
                .build();
    }

}
